package com.minelx.datamicroscope.core.query;

import com.minelx.datamicroscope.core.condition.KeyOperandCondition;
import com.minelx.datamicroscope.core.values.CommonField;
import com.minelx.hellopandas.core.DataFrame;
import com.minelx.hellopandas.core.DataFrameDescription;
import com.minelx.hellopandas.core.Pandas;
import com.minelx.sqlstructures.statements.SelectBuilder;
import com.minelx.sqlstructures.statements.clause.Field;
import com.minelx.sqlstructures.statements.clause.Sql;
import com.minelx.statementsplus.cha.base.ChaSession;
import com.minelx.statementsplus.cha.base.TaskCommitment;
import io.gitee.minelx.filesystem.File;

import java.util.List;
import java.util.Map;

import static com.minelx.datamicroscope.core.condition.Operand.gt;
import static java.lang.String.format;
import static java.util.Collections.singletonList;
import static java.util.stream.Collectors.joining;
import static java.util.stream.Collectors.toList;

public class TrickleQuery {

	private final SelectBuilder builder;

	private final List<AliasedKey> queriedAliasedKeys;

	private final CommonField primaryField;

	private final String databaseIndex;

	private final boolean tryingToGetPrimaryField;

	TrickleQuery(SelectBuilder builder,
	             List<AliasedKey> queriedAliasedKeys,
	             CommonField primaryField,
	             String databaseIndex,
	             boolean tryingToGetPrimaryField) {
		this.builder = builder;
		this.queriedAliasedKeys = queriedAliasedKeys;
		this.primaryField = primaryField;
		this.databaseIndex = databaseIndex;
		this.tryingToGetPrimaryField = tryingToGetPrimaryField;
	}

	public String representTitleExceptPrimary() {
		return String.join(",", representColumnsExceptPrimary0());
	}

	public List<String> representColumnsExceptPrimary0() {
		return builder.select().getFields().stream()
				.filter(field -> !field.equals(primaryField.toSqlField()))
				.map(Field::representName)
				.collect(toList());
	}

	public String representTaskAsTitle() {
		return builder.select().getFields().stream()
				.map(Field::representName)
				.collect(joining(","));
	}

	String toSQLText() {
		return new EnrichedTaskCommitment(primaryField, databaseIndex, builder).toSQLText();
	}

	TaskCommitment commonTaskCommitment() {
		return new EnrichedTaskCommitment(primaryField, databaseIndex, builder).commonTaskCommitment();
	}

	TaskCommitment wrappedWithSelectBlock(TaskCommitment commonTaskCommitment) {
		// wrap with SELECT * block
		SelectBuilder body = builder.copy();

		body.where().remove(KeyOperandCondition.of(primaryField, gt("{{BVAL}}")).sqlText());

		body.orderBy(Sql.empty());
		body.limit(Sql.empty());

		String innerSql = body.build().toText();

		return new TaskCommitment(commonTaskCommitment.loopId(),
				format("%s=> " +
								"SELECT * FROM (%s) a " +
								"WHERE a.%s > {{BVAL}} " +
								"ORDER BY a.%s LIMIT 1000;",
						databaseIndex,
						innerSql.substring(0, innerSql.length() - 1),
						primaryField.name(),
						commonTaskCommitment.loopId()),
				commonTaskCommitment.childSql(),
				commonTaskCommitment.additionalValues(),
				commonTaskCommitment.multipleValues());
	}

	public DataFrame doQuery(ChaSession session, File destination) {
		doSimplyQuery(session, destination);

		// make queried data frame type converters

		Map<String, String> dtypes = dtypesOfQueriedContent();

		// FIXME wait for:  remove drop execution of pandasWithOnlyLocalFile()
		DataFrame result = Pandas.of(Query.localPandas())
				.dataFrame(DataFrameDescription.create(destination.path().uri(), dtypes));
		if (!tryingToGetPrimaryField) {
			result.drop(singletonList(primaryField.name()));
		}
		return result;
	}

	public Map<String, String> dtypesOfQueriedContent() {
		return Resource.dtypesFromAliasedKeys(queriedAliasedKeys);
	}

	public void doSimplyQuery(ChaSession session, File destination) {
		TaskCommitment taskCommitment = explainForCommitment(session);
		new SingleQueryJob(session, taskCommitment, representTaskAsTitle()).work(destination);
	}

	public TaskCommitment explainForCommitment(ChaSession session) {
		TaskCommitment commonTaskCommitment = commonTaskCommitment();
		TaskCommitment wrappedWithSelectBlock = wrappedWithSelectBlock(commonTaskCommitment);
		if (builder.orderBy() == Sql.empty()) { // not ordered
			return tryExplainToCommitment(session, commonTaskCommitment, wrappedWithSelectBlock);
		}
		// sql contains order by
		return tryExplainToCommitment(session, wrappedWithSelectBlock, commonTaskCommitment);
	}

	private static TaskCommitment tryExplainToCommitment(ChaSession session,
	                                                     TaskCommitment first,
	                                                     TaskCommitment second) {
		if (session.explainForResponse(first.majorSql()).isSuccess()) {
			return first;
		}
		if (session.explainForResponse(second.majorSql()).isSuccess()) {
			return second;
		}
		throw new IllegalArgumentException("can't pass explanation, first trying: " + first.majorSql() +
				".\n and then the second trying: " + second.majorSql());
	}
}
